package com.heima.behavior.service.impl;

import com.alibaba.fastjson.JSON;
import com.heima.behavior.dto.ReadBehaviorDto;
import com.heima.behavior.entity.ApReadBehavior;
import com.heima.behavior.entity.UpdateArticleMessage;
import com.heima.behavior.service.IReadService;
import com.heima.common.dto.ResponseResult;
import com.heima.common.dto.User;
import com.heima.common.util.AppThreadUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.Date;

@Service
public class ReadServiceImpl implements IReadService {

    @Autowired
    MongoTemplate mongoTemplate;

    @Autowired
    KafkaTemplate<String,String> kafkaTemplate;

    @Value("${topic.read_behavior}")
    private String ReadTopic;

    @Override
    public ResponseResult saveRead(ReadBehaviorDto dto) {

        //构建ApReadBehavior对象
        ApReadBehavior readBehavior = new ApReadBehavior();

        //补充数据
        User user = AppThreadUtil.getThread();
        if (user != null){
            readBehavior.setUserId(user.getUserId());   //用户id
        }else {
            readBehavior.setEquipmentId(dto.getEquipmentId());  //设备id
        }
        readBehavior.setArticleId(dto.getArticleId());  // 文章id
        readBehavior.setCount(1);   //阅读次数  首次默认为1
        readBehavior.setReadDuration(dto.getReadDuration());    //阅读时间单位
        readBehavior.setPercentage(dto.getPercentage());        //阅读百分比
        readBehavior.setLoadDuration(dto.getLoadDuration());    //文章加载时间
        readBehavior.setCreatedTime(new Date());        //创建时间

        //调用mongodb保存用户行为   查询当前用户或者设备对当前文章是否已经有历史记录
        Query query = new Query();
        query.addCriteria(Criteria.where("articleId").is(readBehavior.getArticleId()));
        if (user != null){
            //根据用户id查询
            query.addCriteria(Criteria.where("userId").is(user.getUserId()));
        }else {
            //根据设备id查询
            query.addCriteria(Criteria.where("equipmentId").is(dto.getEquipmentId()));
        }
        ApReadBehavior one = mongoTemplate.findOne(query, ApReadBehavior.class);
        if (one == null){
            //新增
            mongoTemplate.save(readBehavior);
            //发送消息到kafka
            sendMsg(dto);
        }else {
            //有记录   更新 次数+1 时间修改
            Update update = new Update();
            update.inc("count");    //inc具有原子性  对..进行加或减1
            update.set("readDuration",dto.getReadDuration());   //阅读时间
            update.set("percentage",dto.getPercentage());       //阅读百分比
            update.set("loadDuration",dto.getLoadDuration());   //记载时间
            update.set("updatedTime",new Date());   //修改时间
            mongoTemplate.updateFirst(query,update,ApReadBehavior.class);
            //发送消息到kafka
            sendMsg(dto);
        }

        return ResponseResult.okResult();
    }

    /*
        发送消息到kafka
     */
    private void sendMsg(ReadBehaviorDto dto) {
        UpdateArticleMessage message = new UpdateArticleMessage();
        message.setType(1);
        message.setAdd(1);
        message.setArticleId(dto.getArticleId());
        String json = JSON.toJSONString(message);
        kafkaTemplate.send(ReadTopic,json);

    }
}
